#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <sys/wait.h>
#include <sys/time.h>
#include <time.h>
#include <pthread.h>
#include "blocking_queue.h"
#include "protocol.h"


/*
 * globals, initialized once
 * read-only for all client threads
 * */
static char** srcIpsSpoof =  NULL;
static unsigned long numIps = 0;
static int client_time = 10;
static char sock_path[108];

/**
 * reads the next ip from a given file
 * @param file the file to read from
 * @param ip where to store the read ip
 * @return 1 on success, 0 otherwise
 */
static int getIp(FILE *file, char* ip) {
	if(fscanf(file, "%s", ip) == 1){
		return 1;
	}
	return 0;
}
/**
 * reads ips from file and stores them into a local data structure
 */
static void readIps(char* filename) {
	FILE *file;
	if ( (file = fopen(filename, "r")) == NULL) {
		fprintf(stderr,"client:File %s could not be opened.\n", filename);
		exit(EXIT_FAILURE);
	}
	fprintf(stderr,"client:reading file...");
	fflush(stdout);
	numIps = 0;
	char srcIp[INET_ADDRSTRLEN];
	while ((getIp(file, &srcIp)) && !feof(file)) {
		numIps++;
	}
	rewind(file);
	srcIpsSpoof = (char**)malloc(numIps * sizeof(char*));
	unsigned long cnt = 0;
	while (!feof(file) && (getIp(file, &srcIp))) {
		char* entry = (char*)malloc(INET_ADDRSTRLEN);
		strcpy(entry, srcIp);
		srcIpsSpoof[cnt++] = entry;
		//printf("ip: %s\n", entry);
	}
	fclose(file);
	fprintf(stderr,"client:done, number of ips : %ld \n", numIps);
}

static void stress_balance(void* arg){
	queue_t* queue = arg;
	int message_size = sizeof(message_t);

	srand(time(NULL)*getpid());
	struct timeval startTime, endTime;
	gettimeofday(&startTime, NULL);

	int stop = 0;

	long request_num = 0;
	unsigned long timeouts = 0;
	unsigned long tsent = 0;

	while(!stop){
		message_t* message = malloc(message_size);
		memset(message->body, 'a', BODY_SIZE);
		message->body[BODY_SIZE-1] = '\0';
		char* src_ip = srcIpsSpoof[rand()%numIps];
		strcpy(message->id, src_ip);

		if(queue_put(queue, message) == 0){
			tsent+= message_size;
			request_num++;
		}else{
			timeouts++;
		}
		gettimeofday(&endTime, NULL);
		int delay = (endTime.tv_sec - startTime.tv_sec);
		if(delay > client_time){
			stop = 1;
		}

	}
	fprintf(stderr,"\nclient: requests:%ld:tsent:%ld:timeouts:%ld\n", request_num,tsent,timeouts);

	pthread_exit(0);
}

void* client_init(void* params){
	void** client_params = params;
	int client_threads = *((int*)client_params[0]);
	char* ip_file = (char*)client_params[1];
	int a_client_time = *((int*)client_params[2]);
	queue_t* queue = (queue_t*)client_params[3];

	readIps(ip_file);
	client_time = a_client_time;
	fprintf(stderr, "client:clients will run for %d seconds \n", client_time);

	pthread_t* threads = malloc(client_threads * sizeof(pthread_t));

	pthread_attr_t attr;
	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

	int i = 0;
	for (i = 0; i < client_threads; i++) {
		if (pthread_create(&(threads[i]), &attr, stress_balance, queue) != 0) {
			perror("client:Error creating thread\n");
			exit(-1);
		}
	}

	void * status;
	for (i = 0; i < client_threads; i++) {
		if (pthread_join(threads[i], &status) != 0) {
			perror("client:Error joining thread\n");
			exit(-1);
		}
	}
	balance_stop = 1;
	pthread_attr_destroy(&attr);
	free(threads);
	pthread_exit(0);
}
